package com.bite.messageQueue.mqServer.core;

import com.bite.messageQueue.common.Consumer;
import com.bite.messageQueue.common.ConsumerEnv;
import com.bite.messageQueue.common.MqException;
import com.bite.messageQueue.mqServer.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

@Slf4j
public class ConsumerManager {
    //持有上层VirtualHost的引用，用来操作数据
    private VirtualHost parent;
    //指定一个线程池，执行具体回调
    private ExecutorService executor = Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //扫描线程
    private Thread scanThread = null;


    public ConsumerManager(VirtualHost p) {
        parent = p;

        scanThread = new Thread(() -> {
            while (true) {
                try {
                    //取令牌
                    String take = tokenQueue.take();
                    //根据令牌，找到对应的队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(take);
                    if (queue == null) {
                        log.error("队列：" + take + "不存在");
                        throw new MqException("队列：" + take + "不存在");
                    }
                    //消费信息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        scanThread.setDaemon(true);
        scanThread.start();
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoACK, Consumer consumer) throws InterruptedException {
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            log.error("队列：" + queueName + "不存在");
            throw new RuntimeException("队列：" + queueName + "不存在");
        }
        ConsumerEnv env = new ConsumerEnv(consumerTag, queueName, autoACK, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(env);
            //如果当前队列已经有消息了
            int messageSize = parent.getMemoryDataCenter().getMessageSize(queueName);
            for (int i = 0; i < messageSize; i++) {
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        //按轮询方式取出消费者
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if (consumerEnv == null) {
            return;
        }
        //取消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            return;
        }
        //执行回调
        executor.submit(() -> {
            try{
                //把要消费的消息放入待确认集合队列中
                parent.getMemoryDataCenter().addMessageWaitACK(queue.getName(), message);
                //回调操作
                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                log.info("消费成功");
                //判断消息的应答方式
                //自动应答
                if (consumerEnv.isAutoACK()) {
                    //删硬盘
                    if(message.getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    //删待确认队列
                    parent.getMemoryDataCenter().removeMessageWaitACK(queue.getName(), message.getMessageID());
                    //删内存
                    parent.getMemoryDataCenter().removeMessage(message.getMessageID());
                    log.info("消息被自动应答的方式消费");
                }
            } catch (Exception e){
                e.printStackTrace();
            }
        } );
    }
}
